package com.echoplots.kafka.listener;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import com.echoplots.kafka.entity.Message;

@Component
public class KafkaMessageListener {

    private Logger logger = LoggerFactory.getLogger(this.getClass());
    //@KafkaListener(topics = "topic1, topic2") 同时监听来自多个Topic的消息
    //@Header注解来获取当前消息来自哪个分区（partitions）
/*    @KafkaListener(topics = "test", groupId = "test-consumer")
    public void listen( String message,
            @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
        //logger.info("接收消息: {}", message);
    	 logger.info("接收消息: {}，partition：{}", message, partition);
    }*/
    
    
    //@KafkaListener来指定只接收来自特定分区的消息
/*    @KafkaListener(groupId = "test-consumer",
            topicPartitions = @TopicPartition(topic = "test",
                    partitionOffsets = {
                            @PartitionOffset(partition = "0", initialOffset = "0")
                }))
    public void listen(@Payload String message,
                       @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
        logger.info("接收消息: {}，partition：{}", message, partition);
    }*/
    
    
    @KafkaListener(topics = "message", groupId = "message-consumer")
    public void listen(Message message) {
        logger.info("接收消息: {}", message);
    }
}
